/*
 * Copyright (C) 2018 The Asiainfo-Ability Authors
 *
 *      Licensed under the Apache License, Version 2.0 (the "License");
 *      you may not use this file except in compliance with the License.
 *      You may obtain a copy of the License at
 *
 *          http://www.apache.org/licenses/LICENSE-2.0
 *
 *      Unless required by applicable law or agreed to in writing, software
 *      distributed under the License is distributed on an "AS IS" BASIS,
 *      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *      See the License for the specific language governing permissions and
 *      limitations under the License.
 */

package org.asiainfo.ability.base.async;

import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.EventTranslatorOneArg;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import org.junit.internal.runners.statements.RunAfters;

import java.util.concurrent.*;

@SuppressWarnings("unchecked")
public class CustomQueue
{
    private static class ObjectEvent
    {
        private Object value;
        public void set(Object value)
        {
            this.value = value;
        }
        public Object get() {
            return value;
        }
    }

    private static class ObjectEventFactory implements EventFactory<ObjectEvent> {
        public ObjectEvent newInstance()
        {
            return new ObjectEvent();
        }
    }

    private Disruptor<ObjectEvent> disruptor;
    private RingBuffer<ObjectEvent> ringBuffer;

    public static  class ObjectEventHandler implements EventHandler<ObjectEvent>
    {
        private Future future;
        public ObjectEventHandler(Future operation) {
            this.future = operation;
        }
        public void onEvent(final ObjectEvent event, long sequence, boolean endOfBatch)
        {
            future.callback(event.get());
        }
    }

    public interface Future<T> {
        public void callback(T value);
    }
    public CustomQueue(int bufferSize, Future f) {
        ObjectEventFactory factory = new ObjectEventFactory();
        ThreadFactory threadFactory = Executors.defaultThreadFactory();
        disruptor = new Disruptor<ObjectEvent>(factory, bufferSize, threadFactory);
        disruptor.handleEventsWith(new ObjectEventHandler(f));
        disruptor.start();
        ringBuffer = disruptor.getRingBuffer();
    }

    public void publish(Object value) {
        final long sequence = ringBuffer.next();
        try
        {
            ringBuffer.get(sequence).set(value);
        }
        finally
        {
            ringBuffer.publish(sequence);
        }

    }
    public void release() {
        disruptor.shutdown();
    }
    public static void main(String[] args) throws Exception
    {
        CustomQueue customQueue = new CustomQueue(32768, new Future() {
            @Override
            public void callback(Object value)
            {
                System.out.println(value);
                try {
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        customQueue.publish("aaaaaa");
        customQueue.publish("bbbb");
    }
}
